﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using MessageServer.config;
using System.Data.SqlClient;
using MessageServer.serialize;
using System.Reflection;
using MessageServer.pool;
using System.Threading;
using System.Data;
using MessageServer.Messages;

namespace MessageServer.persistence
{
    //different states in which the message can exist
    public enum messagestatus { unacknowledged = 0, acknowledged = 1, processed = 2, processing = 3, sending = 4, sent = 5, invalid = 100 }

    /**
     * Default constructor creator to deserialize serialized and persisted objects. All
     * persisted objects should have default constructors.
     * */
    public class ObjectCreator : ObjectConstructor
    {
        //invoke the default constructor and return an object of type t
        public object construct(Type t)
        {
            ConstructorInfo construct = t.GetConstructor(new Type[0]);
            return construct.Invoke(new object[0]);
        }
    }

    /**
     * An object representation of the persisted message. 
     * */
    public class PersistedMessage
    {
        int _messageId;                 //representation of the id that recognizes the message
        String _remoteAddress;          //the remote address of the client which sent this message
        byte[] _message;                //serialized message to store
        messagestatus _status;          //status of the message
        String _messageType;            //the type of object that was serialized
        Object _object;                 //the object that is serialized or deserialized
        int _receivedMID;               //message id of the message for which this is a reply
        DateTime _sentDateTime;         //time when this message was sent


        /**
         * Retrieves an Id representation for the object to store into the database
         * */
        internal static int Id(Object obj)
        {
            return obj.GetHashCode();
        }

        /**
         * Create an PersistedMessage that will be stored into the database for the given object obj,
         * address and in the format given
         * */
        internal PersistedMessage(Object obj, String address, SerializeFormat format)
        {
            _status = messagestatus.unacknowledged;
            _messageId = Id(obj);
            _message = format.serialize(obj);
            _remoteAddress = address;
            _messageType = obj.GetType().FullName;
            _object = obj;
        }

        internal PersistedMessage(Object replyfor, Object reply, String address, SerializeFormat format)
        {
            _status = messagestatus.sending;
            _messageId = reply.GetHashCode();
            _receivedMID = replyfor.GetHashCode();
            _remoteAddress = address;
            _messageType = reply.GetType().FullName;
            _sentDateTime = DateTime.Now;
            _object = reply;
            _message = format.serialize(reply);
        }

        /**
         * Create a PersistedMessage from the database reader and deserialize it
         * using the format passed in.
         * */
        internal PersistedMessage(SqlDataReader reader, SerializeFormat format)
        {
            int cnt = 0;
            _messageId = (int)reader[cnt]; cnt++;
            _remoteAddress = (String)reader[cnt]; cnt++;
            _message = (byte[])reader[cnt]; cnt++;
            _messageType = (String)reader[cnt]; cnt++;
            _status = (messagestatus)reader[cnt]; cnt++;
            if (reader.FieldCount > cnt) //if received message id is present, retrieve
            {
                _receivedMID = (int)reader[cnt]; cnt++;
            }
            int deserializedlen = 0;
            _object = format.deserialize(_message, Type.GetType(_messageType), new ObjectCreator(), ref deserializedlen);
        }

        //accessors for the various fields
        public int messageId { get { return _messageId; } }
        public String remoteAddress { get { return _remoteAddress; } }
        public Object message { get { return _object; } }
        internal byte[] messagebytes { get { return _message; } }
        public String messageType { get { return _messageType; } }
        public messagestatus status { get { return _status; } }
        public int receivedMessageId { get { return _receivedMID; } }
        public DateTime sentTime { get { return _sentDateTime; } }
    }

    /**
     * This class will currently not use NHibernate, since the features of
     * NHibernate are not really required here. If such a requirement comes in 
     * later, this file can be changed appropriately.
     * 
     * drop table MESSAGE_RECEIVED
     * CREATE TABLE MESSAGE_RECEIVED
     * (
     * 	    MESSAGE_ID INTEGER PRIMARY KEY, //This has to be a hash value, not an auto increment
     * 	    REMOTE_ADDRESS VARCHAR(50) NOT NULL,
     * 	    MESSAGE_CONTENT BINARY(4096) NOT NULL,
     * 	    MESSAGE_TYPE VARCHAR(1024) NOT NULL,
     * 	    RECEIVED_DATETIME DATETIME NOT NULL DEFAULT GETDATE(),
     * 	    MESSAGE_STATUS INTEGER NOT NULL DEFAULT 0,
     * 	    LISTENERS_RUN   VARCHAR(1024) //will be used for recovery
     * 	)
     * 	
     * CREATE TABLE dbo.MESSAGE_SENT
     * (
     * MESSAGE_ID int NOT NULL,
     * REMOTE_ADDRESS varchar(50) NOT NULL,
     * MESSAGE_CONTENT varbinary(4096) NOT NULL,
     * MESSAGE_TYPE varchar(1024) NOT NULL,
     * SEND_STATUS int NOT NULL,
     * CREATED_DATETIME datetime NOT NULL,
     * SENT_DATETIME datetime NULL,
     * RECEIVED_MESSAGE_ID int NOT NULL
     * ) 
     * 	Expects the above as the definition for the table.
     * 	No transactions here since it has only one statement executed.
     * */
    class PersistMessage : IPoolableObject, IDisposable
    {
        //SQL used to persist the message
        const String PERSIST_MSG = "INSERT INTO MESSAGE_RECEIVED (MESSAGE_ID, REMOTE_ADDRESS, MESSAGE_CONTENT, MESSAGE_TYPE) " +
            " VALUES (@msgid, @remoteaddr, @content, @msgtype)";

        const String PERSIST_SEND_MSG = "INSERT INTO MESSAGE_SENT (MESSAGE_ID, RECEIVED_MESSAGE_ID, REMOTE_ADDRESS, MESSAGE_CONTENT, MESSAGE_TYPE, SEND_STATUS) " +
            " VALUES (@msgid, @receivedmsgid, @remoteaddr, @content, @msgtype, @status)";

        //SQL used to update the status of the message
        const String UPDATE_STATUS = "UPDATE MESSAGE_RECEIVED SET MESSAGE_STATUS = @status WHERE MESSAGE_ID = @msgid";

        //SQL used to update the status of the message
        const String UPDATE_SENT_STATUS = "UPDATE MESSAGE_SENT SET SENT_DATETIME = getDate(), SEND_STATUS = @status WHERE MESSAGE_ID = @msgid";

        //SQL used to retrieve messages in a certain status
        const String RETRIEVE_MSGS = "SELECT MESSAGE_ID, REMOTE_ADDRESS, MESSAGE_CONTENT, MESSAGE_TYPE, MESSAGE_STATUS FROM MESSAGE_RECEIVED WHERE MESSAGE_STATUS = @status AND REMOTE_ADDRESS = @remote";

        //SQL used to retrieve messages in a certain status
        const String RETRIEVE_SEND_MSGS = "SELECT MESSAGE_ID, REMOTE_ADDRESS, MESSAGE_CONTENT, MESSAGE_TYPE, SEND_STATUS, RECEIVED_MESSAGE_ID FROM MESSAGE_SENT WHERE SEND_STATUS = @status AND REMOTE_ADDRESS = @remote";

        //SQL used to find the status of a given message
        const String FIND_STATUS = "SELECT MESSAGE_STATUS FROM MESSAGE_RECEIVED WHERE MESSAGE_ID = @msgid";

        //SQL used to find the status of a given message
        const String FIND_SEND_STATUS = "SELECT SEND_STATUS FROM MESSAGE_SENT WHERE MESSAGE_ID = @msgid";

        MessageServerConfig _config;            //The configuration 
        TuningParams _params;                   //The tuning parameters affects performance
        SqlConnection _connection;              //this shd be a pool, hopefully.
        //TODO: should move this interface out of BNG.
        SerializeFormat _serializer;
        SqlCommand _insert;
        SqlCommand _update;
        SqlCommand _select;
        SqlCommand _findStat;
        SqlCommand _insertSent;
        SqlCommand _updateSent;
        SqlCommand _selectSent;
        SqlCommand _findSentStat;
        bool _valid;

        //if this is a valid object and connection was established correctly
        public bool isvalid { get { return _valid; } }


        public PersistMessage(MessageServerConfig conf, TuningParams parms)
        {
            _valid = true;
            _config = conf;
            _params = parms;
            //Cache this once. Can we run this in multiple threads?
            _connection = new SqlConnection(_config.connectionString);
            _connection.Open();
            setupinsert();
            setupupdate();
            setupselect();
            setupfind();
            setupInsertSent();
            setupUpdateSent();
            setupSelectSent();
            setupSentfind();

            //TODO: can be made common between the BNG and here. But.
            Type stype = Type.GetType(_config.persistenceFormat);
            ConstructorInfo construct = stype.GetConstructor(new Type[0]);
            _serializer = (SerializeFormat)construct.Invoke(new object[0]);
        }

        private void setupinsert()
        {
            _insert = new SqlCommand(PERSIST_MSG, _connection);
            SqlParameter parm = new SqlParameter("@msgid", SqlDbType.Int);
            _insert.Parameters.Add(parm);
            parm = new SqlParameter("@remoteaddr", SqlDbType.VarChar, 50);
            _insert.Parameters.Add(parm);
            parm = new SqlParameter("@content", SqlDbType.VarBinary, 4096);
            _insert.Parameters.Add(parm);
            parm = new SqlParameter("@msgtype", SqlDbType.VarChar, 1024);
            _insert.Parameters.Add(parm);
        }

        private void setupupdate()
        {
            _update = new SqlCommand(UPDATE_STATUS, _connection);
            SqlParameter parm = new SqlParameter("@status", SqlDbType.Int);
            _update.Parameters.Add(parm);
            parm = new SqlParameter("@msgid", SqlDbType.Int);
            _update.Parameters.Add(parm);
        }

        private void setupselect()
        {
            _select = new SqlCommand(RETRIEVE_MSGS, _connection);
            SqlParameter parm = new SqlParameter("@status", SqlDbType.Int);
            _select.Parameters.Add(parm);
            parm = new SqlParameter("@remote", SqlDbType.VarChar, 50);
            _select.Parameters.Add(parm);
        }

        private void setupfind()
        {
            _findStat = new SqlCommand(FIND_STATUS, _connection);
            SqlParameter parm = new SqlParameter("@msgid", SqlDbType.Int);
            _findStat.Parameters.Add(parm);
        }

        private void setupInsertSent()
        {
            _insertSent = new SqlCommand(PERSIST_SEND_MSG, _connection);
            SqlParameter parm = new SqlParameter("@msgid", SqlDbType.Int);
            _insertSent.Parameters.Add(parm);
            parm = new SqlParameter("@receivedmsgid", SqlDbType.Int);
            _insertSent.Parameters.Add(parm);
            parm = new SqlParameter("@remoteaddr", SqlDbType.VarChar, 50);
            _insertSent.Parameters.Add(parm);
            parm = new SqlParameter("@content", SqlDbType.VarBinary, 4096);
            _insertSent.Parameters.Add(parm);
            parm = new SqlParameter("@msgtype", SqlDbType.VarChar, 1024);
            _insertSent.Parameters.Add(parm);
            parm = new SqlParameter("@status", SqlDbType.Int);
            _insertSent.Parameters.Add(parm);
        }

        private void setupUpdateSent()
        {
            _updateSent = new SqlCommand(UPDATE_SENT_STATUS, _connection);
            SqlParameter parm = new SqlParameter("@status", SqlDbType.Int);
            _updateSent.Parameters.Add(parm);
            parm = new SqlParameter("@msgid", SqlDbType.Int);
            _updateSent.Parameters.Add(parm);
        }

        private void setupSelectSent()
        {
            _selectSent = new SqlCommand(RETRIEVE_SEND_MSGS, _connection);
            SqlParameter parm = new SqlParameter("@status", SqlDbType.Int);
            _selectSent.Parameters.Add(parm);
            parm = new SqlParameter("@remote", SqlDbType.VarChar, 50);
            _selectSent.Parameters.Add(parm);
        }

        private void setupSentfind()
        {
            _findSentStat = new SqlCommand(FIND_SEND_STATUS, _connection);
            SqlParameter parm = new SqlParameter("@msgid", SqlDbType.Int);
            _findSentStat.Parameters.Add(parm);
        }

        public override void Dispose()
        {
            if (_insert != null) _insert.Dispose();
            if (_update != null) _update.Dispose();
            if (_select != null) _select.Dispose();
            if (_findStat != null) _findStat.Dispose();
            if (_connection != null) _connection.Close();
        }

        public int persistMessage(Object obj, String address)
        {
            //assumption is that all messages provide the correct implementation
            //of hashcode to compute unique values
            int ret = 0;
            if (_valid)
            {
                //if the message is a persistable message only then try to persist
                if (obj.GetType().IsDefined(typeof(MsgPersistable), false))
                {
                    messagestatus state = messagestate(obj);
                    if (state == messagestatus.invalid)
                    {
                        //insert only if there is no message already present with this hashcode
                        PersistedMessage pm = new PersistedMessage(obj, address, _serializer);

                        int cnt = 0;
                        _insert.Parameters[cnt].Value = pm.messageId; cnt++;
                        _insert.Parameters[cnt].Value = pm.remoteAddress; cnt++;
                        _insert.Parameters[cnt].Value = pm.messagebytes; cnt++;
                        _insert.Parameters[cnt].Value = pm.messageType; cnt++;

                        ret = _insert.ExecuteNonQuery();
                    }
                }
            }

            return ret;
        }

        private void updateStatus(Object obj, messagestatus state)
        {
            if (_valid)
            {
                if (obj.GetType().IsDefined(typeof(MsgPersistable), false))
                {
                    int cnt = 0;
                    int id = PersistedMessage.Id(obj);
                    _update.Parameters[cnt].Value = (int)state; cnt++;
                    _update.Parameters[cnt].Value = id; cnt++;
                    int ret = _update.ExecuteNonQuery();
                    if (ret <= 0) throw new Exception("Did not find object");
                }
            }
        }

        public void processing(Object obj)
        {
            updateStatus(obj, messagestatus.processing);
        }

        public void acknowledged(Object obj)
        {
            updateStatus(obj, messagestatus.acknowledged);
        }

        public void processed(Object obj)
        {
            updateStatus(obj, messagestatus.processed);
        }

        public int persistSend(Object obj, Object received, String address)
        {
            //assumption is that all messages provide the correct implementation
            //of hashcode to compute unique values
            int ret = 0;
            if (_valid)
            {
                //if the message is a persistable message only then try to persist
                if (obj.GetType().IsDefined(typeof(MsgPersistable), false))
                {
                    messagestatus state = sentstate(obj);
                    if (state == messagestatus.invalid)
                    {
                        //insert only if there is no message already present with this hashcode
                        PersistedMessage pm = new PersistedMessage(received, obj, address, _serializer);

                        int cnt = 0;
                        _insertSent.Parameters[cnt].Value = pm.messageId; cnt++;
                        _insertSent.Parameters[cnt].Value = pm.receivedMessageId; cnt++;
                        _insertSent.Parameters[cnt].Value = pm.remoteAddress; cnt++;
                        _insertSent.Parameters[cnt].Value = pm.messagebytes; cnt++;
                        _insertSent.Parameters[cnt].Value = pm.messageType; cnt++;
                        _insertSent.Parameters[cnt].Value = messagestatus.sending; cnt++;

                        ret = _insertSent.ExecuteNonQuery();
                    }
                }
            }

            return ret;
        }

        private void updateSent(Object obj, messagestatus state)
        {
            if (_valid)
            {
                if (obj.GetType().IsDefined(typeof(MsgPersistable), false))
                {
                    int cnt = 0;
                    int id = PersistedMessage.Id(obj);
                    _updateSent.Parameters[cnt].Value = (int)state; cnt++;
                    _updateSent.Parameters[cnt].Value = id; cnt++;
                    int ret = _updateSent.ExecuteNonQuery();
                    if (ret <= 0) throw new Exception("Did not find object");
                }
            }
        }

        public void sent(Object obj)
        {
            updateSent(obj, messagestatus.sent);
        }

        private List<PersistedMessage> getMessages(messagestatus state, String remote)
        {
            List<PersistedMessage> msgs = new List<PersistedMessage>();
            if (_valid)
            {
                int cnt = 0;
                _select.Parameters[cnt].Value = (int)state; cnt++;
                _select.Parameters[cnt].Value = remote; cnt++;
                SqlDataReader reader = _select.ExecuteReader();
                while (reader.Read())
                {
                    PersistedMessage pm = new PersistedMessage(reader, _serializer);
                    msgs.Add(pm);
                }
                reader.Close();
            }
            return msgs;
        }

        public List<PersistedMessage> ackMessages(String remote)
        {
            return getMessages(messagestatus.acknowledged, remote);
        }

        public List<PersistedMessage> processingMessages(String remote)
        {
            return getMessages(messagestatus.processing, remote);
        }

        public List<PersistedMessage> unackMessages(String remote)
        {
            return getMessages(messagestatus.unacknowledged, remote);
        }

        private List<PersistedMessage> getSentMessages(messagestatus state, String remote)
        {
            List<PersistedMessage> msgs = new List<PersistedMessage>();
            if (_valid)
            {
                int cnt = 0;
                _selectSent.Parameters[cnt].Value = (int)state; cnt++;
                _selectSent.Parameters[cnt].Value = remote; cnt++;
                SqlDataReader reader = _select.ExecuteReader();
                while (reader.Read())
                {
                    PersistedMessage pm = new PersistedMessage(reader, _serializer);
                    msgs.Add(pm);
                }
                reader.Close();
            }
            return msgs;
        }

        public List<PersistedMessage> sendingMessages(String remote)
        {
            return getSentMessages(messagestatus.sending, remote);
        }

        public messagestatus messagestate(Object obj)
        {
            messagestatus state = messagestatus.invalid;
            if (_valid)
            {
                if (obj.GetType().IsDefined(typeof(MsgPersistable), false))
                {
                    int cnt = 0;
                    int id = PersistedMessage.Id(obj);
                    _findStat.Parameters[cnt].Value = id;
                    SqlDataReader reader = _findStat.ExecuteReader();
                    if (reader.Read()) //expect only one row.
                    {
                        state = (messagestatus)reader[0];
                    }
                    reader.Close();
                }
            }
            return state;
        }

        public messagestatus sentstate(Object obj)
        {
            messagestatus state = messagestatus.invalid;
            if (_valid)
            {
                if (obj.GetType().IsDefined(typeof(MsgPersistable), false))
                {
                    int cnt = 0;
                    int id = PersistedMessage.Id(obj);
                    _findSentStat.Parameters[cnt].Value = id;
                    SqlDataReader reader = _findSentStat.ExecuteReader();
                    if (reader.Read()) //expect only one row.
                    {
                        state = (messagestatus)reader[0];
                    }
                    reader.Close();
                }
            }
            return state;
        }

    }

    class PersistObjectCreator : IPoolableObjectCreator
    {
        MessageServerConfig _config;
        TuningParams _parms;

        internal PersistObjectCreator(MessageServerConfig conf, TuningParams parms)
        {
            _config = conf;
            _parms = parms;
        }

        public IPoolableObject create()
        {
            return new PersistMessage(_config, _parms);
        }
    }

    public class PooledPersister
    {
        static PooledPersister PERSISTER = null;
        static object lck = new object();

        public static PooledPersister getInstance(MessageServerConfig config, TuningParams parms)
        {
            if (PERSISTER == null)
            {
                lock (lck)
                {
                    if (PERSISTER == null)
                        PERSISTER = new PooledPersister(config, parms);
                }
            }

            return PERSISTER;
        }

        StrictObjectPool _dbpool;                   //pool of persistence objects. While connection has pool, it does not have a way to access it
        private PooledPersister(MessageServerConfig config, TuningParams parms)
        {
            //TODO: add min/max
            _dbpool = new StrictObjectPool(new PersistObjectCreator(config, parms), 100, 10, 50);
        }

        private PoolObject getPersister()
        {
            PoolObject persister = null;
            while (persister == null)
            {
                Thread.Sleep(100); //check every 100ms, without this we can't continue
                persister = _dbpool.get();
            }

            return persister;
        }

        public int persistObjects(List<Object> msgs, String remoteaddr)
        {
            int ret = 0;
            PoolObject persister = getPersister();
            PersistMessage p = persister.poolobject as PersistMessage;
            foreach (Object msg in msgs)
            {
                ret = p.persistMessage(msg, remoteaddr);
            }

            _dbpool.release(persister);
            return ret;
        }

        public int persistResponse(Object msg, Object replyto, String address)
        {
            //will persist only if there is a reply to, else no persistence.
            int ret = 0;
            if (replyto != null)
            {
                PoolObject persister = getPersister();
                PersistMessage p = persister.poolobject as PersistMessage;
                ret = p.persistSend(msg, replyto, address);
                _dbpool.release(persister);
            }

            return ret;
        }

        public void markProcessing(Object msg)
        {
            PoolObject persister = getPersister();
            PersistMessage p = persister.poolobject as PersistMessage;
            p.processing(msg);
            _dbpool.release(persister);
        }

        public void markProcessed(Object msg)
        {
            PoolObject persister = getPersister();
            PersistMessage p = persister.poolobject as PersistMessage;
            p.processed(msg);
            _dbpool.release(persister);
        }

        public void markSent(Object msg)
        {
            PoolObject persister = getPersister();
            PersistMessage p = persister.poolobject as PersistMessage;
            p.sent(msg);
            _dbpool.release(persister);
        }

        public List<PersistedMessage> messagesToBeSent(String remote)
        {
            PoolObject persister = getPersister();
            PersistMessage p = persister.poolobject as PersistMessage;
            List<PersistedMessage> ret = p.sendingMessages(remote);
            _dbpool.release(persister);
            return ret;
        }

        public List<PersistedMessage> messagesToBeAcknowledged(String remote)
        {
            PoolObject persister = getPersister();
            PersistMessage p = persister.poolobject as PersistMessage;
            List<PersistedMessage> ret = p.unackMessages(remote);
            _dbpool.release(persister);
            return ret;
        }

        public List<PersistedMessage> messagesToBeProcessed(String remote)
        {
            PoolObject persister = getPersister();
            PersistMessage p = persister.poolobject as PersistMessage;
            List<PersistedMessage> ret = p.ackMessages(remote);
            List<PersistedMessage> ret1 = p.processingMessages(remote);
            ret.AddRange(ret1); //for now process all messages
            _dbpool.release(persister);
            return ret;
        }

    }
}
